package com.zxy.ziems.common.mqtt;

import com.zxy.ziems.common.mqtt.pojo.Pattern;
import com.zxy.ziems.common.mqtt.pojo.SubscriptTopic;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;

import java.util.List;

/**
 * mqtt 回调类
 * @author 10348359
 * @create 2024-04-29 14:42
 */
@Slf4j
public class MqttCallbackImpl implements MqttCallbackExtended {

    private final List<SubscriptTopic> topicMap;

    private final MqttClient mqttClient;

    private final MqttConnectOptions option;

    public MqttCallbackImpl(List<SubscriptTopic> topicMap,
                            MqttClient mqttClient,
                            MqttConnectOptions option) {
        this.topicMap = topicMap;
        this.mqttClient = mqttClient;
        this.option = option;
    }

    /**
     * 客户端断开后触发
     *
     * @param throwable 异常
     */
    @SneakyThrows
    @Override
    public void connectionLost(Throwable throwable) {
        Integer max = 10;
        while (!mqttClient.isConnected() && max > 0) {
            max = max -1;
            log.debug("[MQTT]Emqx重新连接");
            mqttClient.connect(option);
        }
    }

    /**
     * 客户端收到消息触发
     *
     * @param topic   主题
     * @param message 消息
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("[MQTT]接收到数据主题：{}", topic);
        for (SubscriptTopic subscriptTopic : topicMap) {
            if (subscriptTopic.getPattern() != Pattern.NONE && isMatched(subscriptTopic.getTopic(), topic)) {
                subscriptTopic.getMessageListener().messageArrived(topic, message);
                break;
            }
        }
    }

    /**
     * 检测一个主题是否为一个通配符表示的子主题
     *
     * @param topicFilter 通配符主题
     * @param topic       子主题
     * @return 是否为通配符主题的子主题
     */
    private boolean isMatched(String topicFilter, String topic) {
        return MqttTopic.isMatched(topicFilter, topic);
    }

    /**
     * 发布消息成功
     *
     * @param token token
     */
    @SneakyThrows
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        String[] topics = token.getTopics();
        for (String topic : topics) {
            log.info("[MQTT]向主题：{}发送数据", topic);
        }
    }

    /**
     * 连接emq服务器后触发
     *
     * @param b
     * @param s
     */
    @SneakyThrows
    @Override
    public void connectComplete(boolean b, String s) {
        if (mqttClient.isConnected()) {
            for (SubscriptTopic sub : topicMap) {
                mqttClient.subscribe(sub.getSubTopic(), sub.getQos(), sub.getMessageListener());
                log.info("[MQTT]订阅主题:" + sub.getSubTopic());
            }
            log.info("[MQTT]共订阅:{}个主题!", topicMap.size());
        }
    }
}

